package com.bml.architect.rocketmq.producer;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.StringMessageConverter;

/**
 * rocketmq的事务监听器
 */
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class TransactionMsg implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        String destination = (String) arg;
        org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(),"UTF-8",destination,msg);
        String tags = message.getTags();
        if(tags.contains("TagA")){
            //返回提交的状态，则这个消息会被下游的消费服务所接收到
            return RocketMQLocalTransactionState.COMMIT;
        }
        if(tags.contains("TagB")){
            //这个消息会被丢弃
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        //当返回这个状态的时候，rocketmq隔一会儿回来检查事务的状态，调用的是checkLocalTransaction方法
        //如果checkLocalTransaction方法返回的事务状态是是提交，那么会被下游的消费服务所接收到
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        //这里就可以去检查事务的状态，比如订单是否支付之类的，如果检查用户已经支付了，那么直接提交事务为COMMIT

        return RocketMQLocalTransactionState.COMMIT;
    }
}
